package com.hins.mq.transactionMq;

import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.concurrent.ConcurrentHashMap;

public class TransactionListenerImpl implements TransactionListener {

    //存储对应事务的状态信息  key：事务id, value: 当前事务状态
    private ConcurrentHashMap<String, Integer> localTrans = new ConcurrentHashMap<String, Integer>();

    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        System.out.println("执行本地事务");
        String transactionId = msg.getTransactionId();
        //业务执行，处理本地事务，service层，这里用标签简单分类，正常应该是执行逻辑代码处理异常，执行事务
        if (StringUtils.equals("TagA", msg.getTags())) {
            localTrans.put(transactionId, 1);
            return LocalTransactionState.COMMIT_MESSAGE;
        } else if (StringUtils.equals("TagB", msg.getTags())) {
            localTrans.put(transactionId, 2);
            return LocalTransactionState.ROLLBACK_MESSAGE;
        } else {
            localTrans.put(transactionId, 0);
            return LocalTransactionState.UNKNOW;
        }

    }

    /**
     * 消息回查
     * @param msg
     * @return
     */
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        System.out.println("MQ检查消息Tag【"+msg.getTags()+"】的本地事务执行结果");
        String transactionId = msg.getTransactionId();

        Integer status = localTrans.get(transactionId);//获取该事务id的状态

        System.out.println("消息回查事务id"+transactionId+"  状态是"+status);

        switch (status){
            case 1:
                return LocalTransactionState.COMMIT_MESSAGE;
            case 2:
                return LocalTransactionState.ROLLBACK_MESSAGE;
            default:
                return LocalTransactionState.UNKNOW;
        }
//        return LocalTransactionState.COMMIT_MESSAGE;
    }
}